package com.sf.iisp.kafka.listener;

import com.sf.kafka.api.consume.*;
import com.sf.kafka.exception.KafkaException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.util.Assert;

public class KafkaConsumeListenerContainer implements InitializingBean {
    private static final Logger logger = LoggerFactory.getLogger(KafkaConsumeListenerContainer.class);

    private IStringMessageConsumeListener sgsListener;

    public IStringMessageConsumeListener getSgsListener() {
        return sgsListener;
    }

    public void setSgsListener(IStringMessageConsumeListener sgsListener) {
        this.sgsListener = sgsListener;
    }

    private int messageConsumeThreadCount = 1;

    public int getMessageConsumeThreadCount() {
        return messageConsumeThreadCount;
    }

    public void setMessageConsumeThreadCount(int messageConsumeThreadCount) {
        this.messageConsumeThreadCount = messageConsumeThreadCount;
    }

    private String systemName;
    private String checkCode;
    private String monitorUrl;
    private String clusterName;
    private String topic;
    private ConsumeOptionalConfig consumeOptionalConfig;
    private boolean autoStartup = true;
    private int startupDelay = 0;


    public void setSystemName(String systemName) {
        this.systemName = systemName;
    }

    public void setCheckCode(String checkCode) {
        this.checkCode = checkCode;
    }

    public void setMonitorUrl(String monitorUrl) {
        this.monitorUrl = monitorUrl;
    }

    public void setClusterName(String clusterName) {
        this.clusterName = clusterName;
    }

    public void setTopic(String topic) {
        this.topic = topic;
    }

    public void setConsumeOptionalConfig(ConsumeOptionalConfig consumeOptionalConfig) {
        this.consumeOptionalConfig = consumeOptionalConfig;
    }

    public void setAutoStartup(boolean autoStartup) {
        this.autoStartup = autoStartup;
    }

    public void setStartupDelay(int startupDelay) {
        this.startupDelay = startupDelay;
    }

    private ConsumeConfig buildConsumeConfig() {
        Assert.notNull(systemName, "systemName can not be null!");
        Assert.notNull(checkCode, "checkCode can not be null!");
        Assert.notNull(monitorUrl, "monitorUrl can not be null!");
        Assert.notNull(clusterName, "clusterName can not be null!");
        Assert.notNull(topic, "topic can not be null!");
        Assert.notNull(sgsListener, "messageConsumeListener can not be null!");
        final String systemIdToken = systemName + ":" + checkCode;
        ConsumeConfig consumeConfig = new ConsumeConfig(systemIdToken, monitorUrl, clusterName, topic, messageConsumeThreadCount);
        return consumeConfig;
    }

    @Override
    public void afterPropertiesSet() throws KafkaException {
        if (autoStartup) {
            Assert.notNull(systemName, "systemName can not be null!");
            Assert.notNull(checkCode, "checkCode can not be null!");
            Assert.notNull(monitorUrl, "monitorUrl can not be null!");
            Assert.notNull(clusterName, "clusterName can not be null!");
            Assert.notNull(topic, "topic can not be null!");
            Assert.notNull(sgsListener, "messageConsumeListener can not be null!");

            final String systemIdToken = systemName + ":" + checkCode;
            ConsumeConfig consumeConfig = new ConsumeConfig(systemIdToken, monitorUrl, clusterName, topic, messageConsumeThreadCount);

            StartupHelper helper = new StartupHelper(consumeConfig, sgsListener);
            startListener(helper, startupDelay);
        } else {
            //autoStartup=false,不启动kafka listener,方便研发！
        }
    }

    private void startListener(final StartupHelper helper, final int startupDelay) throws KafkaException {
        if (startupDelay <= 0) {
            helper.startup();
        } else {
            Thread schedulerThread = new Thread() {
                @Override
                public void run() {
                    try {
                        Thread.sleep(startupDelay * 1000);
                    } catch (InterruptedException ex) {
                    }
                    logger.info("Starting kafka consume listener now, after delay of " + startupDelay + " seconds.");
                    try {
                        helper.startup();
                    } catch (KafkaException ex) {
                        throw new RuntimeException("startup kafka consume listener fail!", ex);
                    }
                }
            };
            schedulerThread.setDaemon(true);
            schedulerThread.start();
        }
    }

    class StartupHelper {
        private ConsumeConfig config;
        private IMessageConsumeListener messageConsumeListener;

        StartupHelper(ConsumeConfig config, IMessageConsumeListener messageConsumeListener) {
            this.config = config;
            this.messageConsumeListener = messageConsumeListener;
        }

        void startup() throws KafkaException {
            if (messageConsumeListener instanceof IStringMessageConsumeListener) {
                KafkaConsumerRegister.registerStringConsumer(config, (IStringMessageConsumeListener) messageConsumeListener, consumeOptionalConfig);
            } else if (messageConsumeListener instanceof IByteArrayMessageConsumeListener) {
                KafkaConsumerRegister.registerByteArrayConsumer(config, (IByteArrayMessageConsumeListener) messageConsumeListener, consumeOptionalConfig);
            } else {
                throw new KafkaException("Unknown listener type!" + messageConsumeListener.toString());
            }
        }
    }

}

